package cn.csg.ams.collect2.poptopics.datahub;


import cn.csg.ams.collect2.poptopics.dao.MkPrPfOutageRecoveryMonDao;
import cn.csg.ams.collect2.poptopics.dataHandle.AlarmHandle;
import cn.csg.ams.collect2.poptopics.dataHandle.LoadHandle;
import cn.csg.ams.collect2.poptopics.dataHandle.MeterReadHandle;
import cn.csg.ams.collect2.poptopics.dataHandle.OutgaeAndRecover;
import cn.csg.ams.collect2.poptopics.dataHandle.PowerHandle;
import cn.csg.ams.collect2.poptopics.entity.OutageJudgeVO;
import cn.csg.ams.collect2.poptopics.redisPool.AlarmPool;
import cn.csg.ams.collect2.poptopics.redisPool.OutagePool;
import cn.csg.ams.collect2.poptopics.service.MkMcTermAlarmLogService;
import cn.csg.ams.collect2.poptopics.service.MkPrPfOutageEventDetailService;
import com.aliyun.datahub.client.model.TupleRecordData;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 消费datahub
 */
@Component
public class DatahubComsumer {

    private static Logger logger = LoggerFactory.getLogger(DatahubComsumer.class);

    @Value("${datahub.endpoint}")
    private String endpoint;

    @Value("${datahub.accessId}")
    private String accessId;

    @Value("${datahub.accessKey}")
    private String accessKey;

    /**
     * 告警datahub配置
     */
    @Value("${datahub.projectName}")
    private String projectName;

    @Value("${datahub.topicName}")
    private String topicName;

    @Value("${datahub.subId}")
    private String subId;

    /**
     * 电流电压datahub配置
     */
    @Value("${datahub.loadProjectName}")
    private String loadProjectName;

    @Value("${datahub.loadTopicName}")
    private String loadTopicName;

    @Value("${datahub.loadSubId}")
    private String loadSubId;


    /**
     * 告警数据处理
     */
    @Autowired
    private AlarmHandle alarmHandle;

    /**
     * 电流电压数据处理
     */
    @Autowired
    private LoadHandle loadHandle;

    /**
     * 表码数据处理
     */
    @Autowired
    private MeterReadHandle meterReadHandle;

    /**
     * 功率数据处理
     */
    @Autowired
    private PowerHandle powerHandle;

    @Autowired
    private AlarmPool alarmPool;

    @Autowired
    private OutagePool outagePool;
    @Autowired
    private OutageJudge outageJudge;

    @Autowired
    private KafkaConsumer<String, String> kafkaConsumer;

    @Autowired
    private MkPrPfOutageEventDetailService outageEventDetailService;

    @Autowired
    private MkMcTermAlarmLogService alarmLogService;

    @Autowired
    private OutgaeAndRecover outgaeAndRecover;

    @Autowired
    private MkPrPfOutageRecoveryMonDao outageRecoveryMonDao;


    // @PostConstruct
    public void test() {
        // outageRecoveryMonDao.recover("1", null, null, "2024-04-01 01:00:00", "2", System.currentTimeMillis());

        // Map<String, String> pbTermAndMeterByTaNo = alarmLogService.getPbTermAndMeterByTaNo("099000479673", "01");

        String taType = "01";
        String taNo = "099000479564";
        String areaCode = "030900";
        Integer lowCustCnt = 100;

        // RecordInfo recordInfo = alarmLogService.getRecordInfoByTerm("1111111119433224", areaCode);
        //
        // String runMeterId = "1111111119433224";
        // String alarmSourCode = "1";
        // // 从停电池中查询是否存在该电表的停电信息 （此处复电告警若为集中器，查不出停电实体，无法复电; 区分终端电表的查询方式）
        // OutagePool.OutageInfo outageInfo = null;
        // if ("1".equals(alarmSourCode)) {
        //     // 若为终端复电告警，查询所属台区是否停电
        //     outageInfo = outagePool.getOutageInfoByTaNo(taNo, areaCode)
        //             .stream().filter(o -> "2".equals(o.getObjectType()) || "3".equals(o.getObjectType()))
        //             .findFirst().orElse(null);
        // } else {
        //     outageInfo = outagePool.getOutageInfo(taNo, runMeterId, areaCode);
        // }
        // if (outageInfo == null) {
        //     return;
        // }
        //
        // // 存在停电记录, 为该电表复电
        // // 停电告警时间
        // String tdTime = outageInfo.getAlarmTime();
        // // 复电告警时间
        // String fdTime = "20240419112515";
        // // 复电告警时间 < 停电告警时间， 丢弃
        // if (DateUtil.strToDate(fdTime, "yyyyMMddHHmmss").getTime() < DateUtil.strToDate(tdTime, "yyyyMMddHHmmss").getTime()) {
        //     return;
        // }
        //
        // // 若为复电，则修改数据库中该条记录， 并删除redis停电告警池中的数据
        // outgaeAndRecover.updateAlarm(runMeterId, alarmSourCode, fdTime, recordInfo, "2", 1712598755000L, areaCode);
        //
        //


        List<AlarmPool.AlarmInfo> alarmList = alarmPool.getAlarmByTaNo(taNo, areaCode);

        List<OutagePool.OutageInfo> outageInfoList = outagePool.getOutageInfoByTaNo(taNo, areaCode);

        // 从停电池中取出台区下所有停电信息
        Set<String> objTypeSet = outageInfoList.stream().map(OutagePool.OutageInfo::getObjectType).collect(Collectors.toSet());

        // 研判参数
        OutageJudgeVO param = new OutageJudgeVO();
        // 获取该台区下所有告警信息, 进行复电研判
        param.setAlarmList(alarmList);
        param.setTaNo(taNo);
        param.setTaType(taType);
        param.setLowCustCnt(lowCustCnt);

        List<String> runMeterIds = alarmList.stream().map(AlarmPool.AlarmInfo::getRunMeterId).collect(Collectors.toList());


        // OutageJudgeVO param1 = new OutageJudgeVO();
        // param1.setTaNo(taNo);
        // param1.setTaType(taType);
        // param1.setLowCustCnt(lowCustCnt);
        // alarmList = alarmList.stream().filter(alarmInfo -> "2".equals(alarmInfo.getAlarmSourCode())).collect(Collectors.toList());
        // param1.setAlarmList(alarmList);
        // List<OutageJudgeData> outageJudgeData = outageJudge.outageJudge(param1);

        // 停电的根节点为台区
        if (objTypeSet.contains("2") || objTypeSet.contains("3")) {
            // 研判台区是否复电
            if (outageJudge.taOutageRestoreJudge(param)) {
                System.out.println("台区复电。。。。。。。。");

                // // 修改台区下所有记录
                // outageEventDetailService.updateOutageEvent(recordInfo, outageEndTime, taNo, null, restoreType, startTime);
                // // 删除停电池中整个台区的数据
                // outagePool.deleteTaOutageInfo(taNo, areaCode);
                // // 删除告警池中整个台区的数据
                // alarmPool.deleteTaAlarmInfo(taNo, areaCode);
            }
            // // 删除停电池中整个台区的数据
            // outagePool.deleteTaOutageInfo(taNo, areaCode);
            // // 删除告警池中整个台区的数据
            // alarmPool.deleteTaAlarmInfo(taNo, areaCode);
        }
    }


    /**
     * 消费告警数据
     */
    // @PostConstruct
    public void consumeAlarm() throws Exception {
        kafkaConsumer.subscribe(Arrays.asList("test"));

        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(10));
            if (records.isEmpty()) {
                logger.info("暂无告警数据");
                continue;
            }

            for (ConsumerRecord<String, String> record : records) {
                TupleRecordData data = TupleRecordDataUtils.kafkaToDatahubValue(record.value());
                alarmHandle.handle(data, record.timestamp());
                // System.out.println("消费到数据 " + record.key() + ":" + record.value());
            }
        }
    }


    /**
     * 消费电流电压数据
     */
    // @PostConstruct
    private void comsume() {
        BaseDatahubConsumer consumer = new BaseDatahubConsumer() {
            @Override
            void dataHandle(TupleRecordData data, long systemTime) {
                // executor.execute(() -> loadHandle.handle(data));
                loadHandle.handle(data);
            }
        };
        consumer.setEndpoint(endpoint);
        consumer.setAccessId(accessId);
        consumer.setAccessKey(accessKey);
        consumer.setProjectName(loadProjectName);
        consumer.setTopicName(loadTopicName);
        consumer.setSubId(loadSubId);
        consumer.start();
    }


    /**
     * 消费表码数据复电
     */
    // @PostConstruct
    private void comsume2() {
        BaseDatahubConsumer consumer = new BaseDatahubConsumer() {
            @Override
            void dataHandle(TupleRecordData data, long systemTime) {
                meterReadHandle.handle(data);
            }
        };
        consumer.setEndpoint(endpoint);
        consumer.setAccessId(accessId);
        consumer.setAccessKey(accessKey);
        consumer.setProjectName(projectName);
        consumer.setTopicName("topic_ods_0300domk_run_meter_read_curve_r");
        consumer.setSubId("1715564528035XQV87");
        consumer.start();
    }

    // topic_ods_0300domk_run_meter_cur_curve 电流电压
    // topic_ods_0300domk_run_meter_power_curve 功率

    /**
     * 消费 功率-中高压 数据复电
     */
    // @PostConstruct
    private void comsume3() {
        BaseDatahubConsumer consumer = new BaseDatahubConsumer() {
            @Override
            void dataHandle(TupleRecordData data, long systemTime) {
                powerHandle.handle(data);
            }
        };
        consumer.setEndpoint(endpoint);
        consumer.setAccessId(accessId);
        consumer.setAccessKey(accessKey);
        consumer.setProjectName(projectName);
        consumer.setTopicName("topic_ods_0300domk_run_meter_power_curve_r");
        consumer.setSubId("171567153422001QFA");
        consumer.start();
    }

    /**
     * 消费 功率-低压 数据复电
     */
    // @PostConstruct
    private void comsume4() {
        BaseDatahubConsumer consumer = new BaseDatahubConsumer() {
            @Override
            void dataHandle(TupleRecordData data, long systemTime) {
                powerHandle.handle(data);
            }
        };
        consumer.setEndpoint(endpoint);
        consumer.setAccessId(accessId);
        consumer.setAccessKey(accessKey);
        consumer.setProjectName(projectName);
        consumer.setTopicName("topic_ods_0300domk_run_meter_power_curve_low_r");
        consumer.setSubId("1715671826863YF0JA");
        consumer.start();
    }

}
